package com.flybear.aqi.client;



import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.threadpool.ThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;


public class ElasticSearchHighLevelClient implements ElasticSearchClient {

  public static final Logger logger = LoggerFactory
      .getLogger(ElasticSearchHighLevelClient.class);

  private HttpHost[] serverAddresses;
  private ElasticSearchEventSerializer serializer;
  private ElasticSearchIndexRequestBuilderFactory indexRequestFactory;
  private BulkProcessor bulkProcessor;

//  private RestClient client;
  private RestHighLevelClient highLevelClient;

  HttpHost[] getServerAddresses() {
    return serverAddresses;
  }


  void setBulkRequestBuilder(BulkProcessor bulkProcessor) {
    this.bulkProcessor = bulkProcessor;
  }

  /**
   * Transport client for external cluster
   *
   * @param hostNames
   * @param clusterName
   * @param serializer
   */
  public ElasticSearchHighLevelClient(String[] hostNames, String clusterName,
      ElasticSearchEventSerializer serializer) {
    configureHostnames(hostNames);
    this.serializer = serializer;
    openClient(clusterName);
  }

  public ElasticSearchHighLevelClient(String[] hostNames, String clusterName,
      ElasticSearchIndexRequestBuilderFactory indexRequest) {
    configureHostnames(hostNames);
    this.indexRequestFactory = indexRequest;
    openClient(clusterName);
  }





  private void configureHostnames(String[] hostNames) {
    logger.warn(Arrays.toString(hostNames));
    serverAddresses = new HttpHost[hostNames.length];
    for (int i = 0; i < hostNames.length; i++) {
      String[] hostPort = hostNames[i].trim().split(":");
      String host = hostPort[0].trim();
      int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim())
              : 9200;
      serverAddresses[i] = new HttpHost(host, port, "http");
    }
  }

  @Override
  public void close() {
    if (highLevelClient != null) {
      try {
        highLevelClient.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
    highLevelClient = null;
    try {
      boolean terminated = bulkProcessor.awaitClose(30L, TimeUnit.SECONDS);
      logger.warn("bulkProcessor [{}] executed closed wait 30 seconds", terminated);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  //https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high-document-index.html
  //IndexRequest
  @Override
  public void addEvent(Map event, String indexNameBuilder,
                       String indexType, long ttlMs) throws Exception {

    IndexRequest one = null;
    if (indexRequestFactory == null) {//json
//      indexRequestBuilder = client
//          .prepareIndex(indexNameBuilder.getIndexName(event), indexType)
//          .setSource(segetContentBuilder(event));
//      Requests.indexRequest()
      one=  new IndexRequest(indexNameBuilder, indexType);
      if (serializer != null)one.source(serializer.getContentBuilder(event), XContentType.JSON);
      else
        one.source(event);
    } else {
      one = indexRequestFactory.createIndexRequest(
          indexNameBuilder, indexType, event);
    }

//    if (ttlMs > 0) {
//      indexRequestBuilder.setTTL(ttlMs);
//    }
    bulkProcessor.add(one);

  }
  @Override
  public void execute() throws Exception {
//    logger.debug("total insert count [{}] -------------------", totalinsertCount.get()*1000/totalinserttime.get());
//    try {
//      BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
//      if (bulkResponse.hasFailures()) {
//        throw new EventDeliveryException(bulkResponse.buildFailureMessage());
//      }
//    } finally {
//      bulkRequestBuilder = client.prepareBulk();
//    }
  }
  private AtomicLong totalinsertCount =new AtomicLong();
  private final AtomicLong totalinserttime = new AtomicLong();
  private void initBuildProcesser(){

//    ThreadPool threadPool = new ThreadPool(Settings.builder()
//        .put("node.name", "ElasticSearchHighLevelClientThreadPoolTypePermittedInSettings")
//        .put("thread_pool.bulk.type", ThreadPool.ThreadPoolType.FIXED.getType())
//        .build());
    BulkProcessor.Listener listener = new BulkProcessor.Listener() {
      @Override
      public void beforeBulk(long executionId, BulkRequest request) {
        int numberOfActions = request.numberOfActions();
//        db.addinserted(numberOfActions);
        totalinsertCount.addAndGet(numberOfActions);
        logger.debug("Executing bulk [{}] with {} requests", executionId, numberOfActions);

//        System.out.println((executionId) + Thread.currentThread().getName()
//            + "Executing bulk with {} requests"+(numberOfActions));
      }

      @Override
      public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
        if (response.hasFailures()) {
          logger.warn("Bulk [{}] executed with failures", executionId);
        } else {
          logger.debug("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
          totalinserttime.addAndGet(response.getTook().getMillis());
          System.out.println((executionId) + Thread.currentThread().getName()
              + "Bulk [{}] completed in {} milliseconds"+(response.getTook().getMillis()));
        }
      }

      @Override
      public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
        logger.error("Failed to execute bulk", failure);
      }
    };
    BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
            (request, bulkListener) -> highLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);

    BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);

//    BulkProcessor.Builder builder =   BulkProcessor.builder(highLevelClient::bulkAsync, listener, threadPool);
    builder.setBulkActions(bulkActions);
    ByteSizeValue bs=null;
    if(bulkSize!= -1){
      bs=new ByteSizeValue(bulkSize, ByteSizeUnit.MB);
    }else{
      bs=new ByteSizeValue(-1, ByteSizeUnit.BYTES);
    }
    builder.setBulkSize(bs);
    builder.setConcurrentRequests(concurrentRequests);
    builder.setFlushInterval(TimeValue.timeValueSeconds(flushInterval));
    builder.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueSeconds(timeValueSeconds), maxNumberOfRetries));

       this.bulkProcessor = builder.build();
  }
  private int bulkActions = -1;
  private long bulkSize=5l;
  private long flushInterval=60l;
  private int concurrentRequests = 1;
  private long timeValueSeconds=30l;
  private int  maxNumberOfRetries=3;
  /**
   * Open client to elaticsearch cluster
   *
   * @param clusterName
   */
  private void openClient(String clusterName) {
    logger.info("Using ElasticSearch hostnames: {} ",
        Arrays.toString(serverAddresses));
    //新的实现方式
//    RestClient lowLevelRestClient = RestClient.builder(serverAddresses).build();
    this.highLevelClient =
        new RestHighLevelClient(RestClient.builder(serverAddresses));
//    if (highLevelClient != null) {
//      try {
//        highLevelClient.close();
//      } catch (IOException e) {
//        e.printStackTrace();
//      }
//    }
//    client = lowLevelRestClient;
    if (bulkProcessor == null) {
      initBuildProcesser();
    }

  }



  @Override
  public void configure(Map<String,String> conf, Map<String, Object> confparam ) {
    //To change body of implemented methods use File | Settings | File Templates.
    if (StringUtils.isNotBlank(conf.get(BULKACTIONS))) {
      this.bulkActions =  Integer.parseInt(conf.get(BULKACTIONS));
    }

    if (StringUtils.isNotBlank(conf.get(FLUSHINTERVAL))) {
      this.flushInterval =  Long.parseLong(conf.get(FLUSHINTERVAL));
    }

    if (StringUtils.isNotBlank(conf.get(CONCURRENTREQUESTS))) {
      this.concurrentRequests =  Integer.parseInt(conf.get(CONCURRENTREQUESTS));
    }
    if (StringUtils.isNotBlank(conf.get(BULKSIZE))) {
      this.bulkSize = Long.parseLong(conf.get(BULKSIZE));
    }
    if(confparam!=null){
//      this.db= (DBObjectEx)confparam.get("dbex");
    }
  }
}
